package club.monkeywood.ad.dmp.tool

import club.monkeywood.ad.dmp.bean.Log
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{SQLContext, SparkSession}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 日志转成parquet文件格式
  *
  * 使用自定义类的方式构建schema信息
  */
object Biz2ParquetV2 {

    def main(args: Array[String]): Unit = {

        // 0 校验参数个数
        //logInputPath:输入数据路径
        //compressionCode:输出文件压缩编码
        //resultOutputPath:输出文件路径
        //numPartition:输入数据重新分区数量
        if (args.length != 4) {
            println(
                """
                  |cn.dmp.tools.Bzip2Parquet
                  |参数：
                  | logInputPath
                  | compressionCode <snappy, gzip, lzo>
                  | resultOutputPath
                  | numPartition
                """.stripMargin)
            sys.exit()
        }

        // 1 接受程序参数
        val Array(logInputPath, compressionCode,resultOutputPath,numPartition) = args

        // 2 创建sparkconf->SparkSession
        val sparkConf = new SparkConf()
        sparkConf.setAppName(s"${this.getClass.getSimpleName}")
        sparkConf.setMaster("local[*]")
        //使用KryoSerializer更快
        // RDD 序列化到磁盘 worker与worker之间的数据传输
        sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
        // 注册自定义类的序列化方式
        //指定parquet压缩编码
        sparkConf.set("spark.sql.parquet.compression.codec", compressionCode)
        sparkConf.registerKryoClasses(Array(classOf[Log]))

        val ss = SparkSession.builder()
          .appName(s"${this.getClass.getSimpleName}")
          .config(sparkConf)
          .getOrCreate()

        // 读取日志文件
        // 对输入数据重新分区，以提高executor的利用率
        val dataLog: RDD[Log] = ss.sparkContext.textFile(logInputPath).repartition(numPartition.toInt)
          //-1:整行全部切分
          .map(line => line.split(",", -1))
          .filter(_.length >= 85)
          .map(arr => Log(arr))

        val dataFrame = ss.createDataFrame(dataLog)

        // 按照省份名称及地市名称对数据进行分区
        dataFrame.write.partitionBy("provincename", "cityname")
            .parquet(resultOutputPath)

        ss.stop()

    }


}
